package com.gitee.cui.nio.reactor;

import com.gitee.cui.nio.NioDemoConfig;
import com.gitee.cui.util.Logger;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 单线程版本Reactor模式实现控制台输入回显功能
 * @author cuiqingqiang
 * @version 1.0
 * @date 2021/10/20 13:18
 **/
public class MultiThreadEchoServerReactor {

    public static void main(String[] args) throws IOException {
        new MultiThreadEchoServerReactor().start();
    }


    /**
     * 多个选择器
     * 第一个选择器负责监听连接事件并转发
     * 第二个选择器服务监听读写事件并转发
     */
    private Selector[] selectors = new Selector[2];

    /**
     * 多个子反应器
     * 两个子反应器，和选择器数量相同，避免多个线程处理一个选择器出线线程切换时上下文同步效率低问题
     */
    private SubReactor[] subReactors;

    /**
     * 服务端socket通道
     */
    private ServerSocketChannel serverSocketChannel;

    private AtomicInteger next = new AtomicInteger(0);

    public MultiThreadEchoServerReactor() throws IOException {
        selectors[0] = Selector.open();
        selectors[1] = Selector.open();
        serverSocketChannel = ServerSocketChannel.open();
        // 服务端监听地址
        InetSocketAddress socketAddress = new InetSocketAddress(NioDemoConfig.SOCKET_SERVER_IP, NioDemoConfig.SOCKET_SERVER_PORT);
        serverSocketChannel.socket().bind(socketAddress);
        Logger.info("服务端已经开始监听：" + socketAddress);
        // 设置为非阻塞通道
        serverSocketChannel.configureBlocking(false);
        // 将通道注册到监听器，并监听accept事件
        SelectionKey selectionKey = serverSocketChannel.register(selectors[0], SelectionKey.OP_ACCEPT);
        // 对selectionkey添加附件
        selectionKey.attach(new AcceptorHarndler());
        // 负责新连接事件分发的子反应器（不处理具体业务，只负责分发）
        SubReactor connectReactor = new SubReactor(selectors[0]);
        // 负责读写事件分发的子反应器（不处理具体业务，只负责分发）
        SubReactor readWriteReactor = new SubReactor(selectors[1]);
        subReactors = new SubReactor[]{connectReactor, readWriteReactor};
    }

    public void start() {
        new Thread(subReactors[0]).start();
        new Thread(subReactors[1]).start();
    }

    private void dispatch(SelectionKey sk) {
        // 是一个线程
        Runnable handler = (Runnable) sk.attachment();
        // 调用之前attach绑定到选择键的handler处理器对象
        if (handler != null) {
            handler.run();
        }
    }

    /**
     * 新连接处理器
     */
    class AcceptorHarndler implements Runnable {

        @Override
        public void run() {
            try {
                SocketChannel socketChannel = serverSocketChannel.accept();
                Logger.info("接收到一个新连接……，处理读写逻辑");
                if (socketChannel != null) {
                    int index = next.get();
                    Logger.info("选择器的编号：" + index);
                    Selector selector = selectors[index];
                    new MultiThreadEchoHandler(selector, socketChannel);
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
            if (next.incrementAndGet() == selectors.length) {
                next.set(0);
            }
        }
    }

    /**
     * 子反应器
     */
    private class SubReactor implements Runnable {

        private Selector selector;

        public SubReactor(Selector selector) {
            this.selector = selector;
        }

        @Override
        public void run() {
            try {
                while (!Thread.interrupted()) {
                    selector.select(1000);
                    Set<SelectionKey> selectionKeys = selector.selectedKeys();
                    if (selectionKeys == null || selectionKeys.isEmpty()) {
                        continue;
                    }
                    Iterator<SelectionKey> iterator = selectionKeys.iterator();
                    while (iterator.hasNext()) {
                        SelectionKey sk = iterator.next();
                        // 只负责dispatch接收到的事件，具体处理过程由单独的处理类进行处理
                        dispatch(sk);
                    }
                    selectionKeys.clear();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}
